#=================================
# 函数定义
#=================================

import pandas as pd
import numpy as np
import os
from datetime import datetime

# 数据处理与标准化函数
def standardize_data(df_m, df_A):
    # 统一日期格式
    df_m['date'] = pd.to_datetime(df_m['date']).dt.strftime('%Y-%m-%d')
    df_A['time'] = pd.to_datetime(df_A['time']).dt.strftime('%Y-%m-%d')
    # 统一股票代码格式
    def replace_code(x):
        if 'szse.' in x:return x.replace('szse.', '') + '.XSHE'
        if 'sse.' in x:return x.replace('sse.', '') + '.XSHE'
        return x
    df_A['code'] = df_A['code'].apply(replace_code)
    # 统一字段名称,只保留需要对比的字段
    df_A.rename(columns={
        'time': 'date',
        'total_turnover': 'amount',
        'code': 'order_book_id'
        }, inplace=True)
    cols = ['date', 'order_book_id', 'open', 'high', 'low', 'close']
    df_m = df_m[cols].copy()
    df_A = df_A[cols].copy()

    # 对所有数值列保留1位小数
    num_cols = ['open', 'high', 'low', 'close']
    df_m[num_cols] = df_m[num_cols].round(1)
    df_A[num_cols] = df_A[num_cols].round(1)
    return df_m, df_A

# 字段一致性检查
def check_field(df):
    # 标记数据来源
    df['数据来源'] = df['_merge'].map({
        'left_only': '仅米筐',
        'right_only': '仅公司',
        'both': '两者共有'
    })
    if (df['数据来源']=='两者共有').sum()!=0:
        return str(df['数据来源'].value_counts())
    else:
        return ""
    
#检查缺失值
def check_null(df):
    text=[]
    H,L = np.where(df.isnull())
    if len(H)!=0:
        text.append(f"发现{len(H)}处缺失值:")
        for h,l in zip(H,L):
            text.append(f"\t时间：{df["date"].iloc[h]}  列名：{df.columns[l]}")
    return "\n".join(text)

# 日期对齐检查
def check_data(df_m,df_A):
    text=[]
    m_date = set(df_m.date)
    A_date = set(df_A.date)
    if len(m_date - A_date)!=0:
        text.append("米筐独有日期:"+str(len(m_date - A_date))+"\n\t"+"\t".join(list(m_date - A_date)))
    if len(A_date - m_date)!=0:
        text.append("公司独有日期:"+str(len(A_date - m_date))+"\n\t"+"\t".join(list(A_date - m_date)))
    return "\n".join(text)

# 检查价格字段差异
def check_diff(df):
    text = []
    for n in ['open', 'high', 'low', 'close']:
        c = (df[f'{n}_x']==df[f'{n}_y'])
        if len(df[c])!=0:
            text.append(f"{n}字段差异: {len(df[c])}处"+"↓"*5)
            text.append(f"日期\t\t\t米筐\t\tAT")
            for a in np.where(c)[0]:
                text.append(f"{df.date.iloc[a]}\t{df[f'{n}_x'].iloc[a]}\t\t{df[f'{n}_y'].iloc[a]}")
        return "\n".join(text)
    
# 保存报告分块函数
def save_report_in_chunks(report, base_filename, max_files=8):
    # 计算总字符数（包括换行符）
    total_chars = sum(len(line) for line in report) + len(report)
    # 计算每个文件的最小字符数
    min_chars_per_file = total_chars // max_files
    # 分块保存
    file_paths = []
    current_chunk = []
    current_char_count = 0
    file_count = 1
    for line in report:
        line_length = len(line) + 1  # 包括换行符
        # 如果当前块不为空且添加新行会超过最小字符数，或者文件数未达到上限
        if current_chunk and (current_char_count + line_length > min_chars_per_file) and file_count < max_files:
            # 保存当前块
            chunk_filename = f"{base_filename}_part{file_count}.txt"
            with open(chunk_filename, 'w', encoding='utf-8') as f:
                f.write("\n".join(current_chunk))
            file_paths.append(chunk_filename)
            # 重置当前块
            current_chunk = []
            current_char_count = 0
            file_count += 1
        # 添加行到当前块
        current_chunk.append(line)
        current_char_count += line_length
    # 保存最后一个块
    if current_chunk:
        chunk_filename = f"{base_filename}_part{file_count}.txt"
        with open(chunk_filename, 'w', encoding='utf-8') as f:
            f.write("\n".join(current_chunk))
        file_paths.append(chunk_filename)
    return file_paths


#=================================
# 主函数
#=================================
# 准备报告内容
report = []
current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
report.append(f"数据质量检验报告 ({current_time})\n")
report.append("="*50 + "\n")

file = r"C:\Users\柔莹\Desktop\点宽培训\数据质检\A股日线数据\米筐的A股日线"
m_path = pd.Series(os.listdir(file), index=[t.split(".")[0] for t in os.listdir(file)],name="MK")
file = r"C:\Users\柔莹\Desktop\点宽培训\数据质检\A股日线数据\AT的A股日线"
A_path = pd.Series(os.listdir(file), index=[t.split(".")[1] for t in os.listdir(file)],name="AT")

# 合并文件
file_path = pd.concat([m_path, A_path],axis=1)
com_f = file_path.dropna()

# 比较两个文件
report.append(f"【股票覆盖情况】")
report.append(f"米筐的A股日线股票数量: {len(m_path)}")
report.append(f"AT的A股日线股票数量: {len(A_path)}")
report.append(f"共同股票数量: {file_path.notnull().all(axis=1).sum()}")
report.append(f"仅米筐所有的股票: {file_path.AT.isnull().sum()}")
report.append(f"仅AT所有的股票: {file_path.MK.isnull().sum()}")

report.append(f"仅米筐所有的股票代码:\n\t {"\t".join(file_path[file_path.AT.isnull()].index)}")

进度=0
总=len(com_f)
for c in com_f.index:
    df_m =pd.read_csv("./A股日线数据/米筐的A股日线/"+com_f.loc[c,"MK"])
    df_A=pd.read_csv("./A股日线数据/AT的A股日线/"+com_f.loc[c,"AT"])
    report.append("-"*50)
    report.append(f"检查股票代码：{c}")
    df_m, df_A = standardize_data(df_m, df_A)
    df_all = pd.merge(df_m,df_A,
            on=['order_book_id', 'date'],
            how='outer',
            indicator=True)
    
    c=check_field(df_all)
    if len(c)!=0:
        report.append("字段不一致差异——————")
        report.append(c)
    c=check_null(df_m)
    if len(c)!=0:
        report.append("米筐数据缺失值检查——————")
        report.append(c)
    c=check_null(df_A)
    if len(c)!=0:
        report.append("AT数据缺失值检查——————")
        report.append(c)
    c=check_data(df_m,df_A)
    if len(c)!=0:
        report.append("日期对齐检查——————")
        report.append(c)
    c=check_diff(df_all)
    if len(c)!=0:
        report.append("检查价格字段差异——————")
        report.append(c)

    进度=进度+1
    if 进度%500==0:print(f"完成进度：{进度}/{总}")
print("完成检查")

# 保存报告
report_path = f"./报告/数据质检报告_{current_time[:10]}"
saved_files = save_report_in_chunks(report, report_path, max_files=8)
print(f"报告已生成: {report_path}")

# 报告示意
for r in report[:20]:
    print(r)



